package chapter10

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, Encoders, Row, SparkSession}

/**
 * author: 余辉
 * blog: https://blog.csdn.net/silentwolfyh
 * descriptions: 1.1.1.2 从RDD[java class类]创建Dataset
 * date: 2024 - 09 - 02 3:39 下午
 *
 * 创建一个javaBean 的RDD
 * 隐式转换中没有支持好对javabean的encoder机制
 * 需要自己传入一个encoder
 * 可以构造一个简单的encoder，具备序列化功能，但是不具备字段解构的功能
 * 但是，至少能够把一个RDD[javabean] 变成一个 dataset[javabean]
 * 后续可以通过rdd的map算子将数据从对象中提取出来，组装成tuple元组，然后toDF即可进入sql空间
 */
object RDDToDS02 {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("")
      .master("local[*]")
      .getOrCreate()

    // 这句很重要，隐士转换
    import spark.implicits._

    val rdd2: RDD[Stu2] = spark.sparkContext.parallelize(Seq(
      new Stu2(1,"a",18,"上海",99.9),
      new Stu2(2,"b",28,"北京",99.9),
      new Stu2(3,"c",38,"西安",99.9)
    ))

    val encoder = Encoders.kryo(classOf[Stu2])
    val ds2: Dataset[Stu2] = spark.createDataset(rdd2)(encoder)

    val df2: Dataset[Row] = ds2.map(stu => {
      (stu.getId, stu.getName, stu.getAge)
    }).toDF("id", "name", "age")
    ds2.printSchema()
    ds2.show()
    df2.show()

  }
}